package com.lesent.kafka.config;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * @author shengtao
 * @Description:
 * @date 2018/08/14 14:30
 */
@Configuration
@ConfigurationProperties(prefix = "kafka.producer")
@EnableKafka
public class Producer {

    private String bootstrapServers;

    private String keySerializer;

    private String valueSerializer;

    private String batchSize;

    private String bufferMemory;

    private int retries;

    private int lineMs;

    private String deafaultTopic;

    public String getBootstrapServers() {
        return bootstrapServers;
    }

    public void setBootstrapServers(String bootstrapServers) {
        this.bootstrapServers = bootstrapServers;
    }

    public String getKeySerializer() {
        return keySerializer;
    }

    public void setKeySerializer(String keySerializer) {
        this.keySerializer = keySerializer;
    }

    public String getValueSerializer() {
        return valueSerializer;
    }

    public void setValueSerializer(String valueSerializer) {
        this.valueSerializer = valueSerializer;
    }

    public String getBatchSize() {
        return batchSize;
    }

    public void setBatchSize(String batchSize) {
        this.batchSize = batchSize;
    }

    public String getBufferMemory() {
        return bufferMemory;
    }

    public void setBufferMemory(String bufferMemory) {
        this.bufferMemory = bufferMemory;
    }

    public int getRetries() {
        return retries;
    }

    public void setRetries(int retries) {
        this.retries = retries;
    }

    public int getLineMs() {
        return lineMs;
    }

    public void setLineMs(int lineMs) {
        this.lineMs = lineMs;
    }

    public String getDeafaultTopic() {
        return deafaultTopic;
    }

    public void setDeafaultTopic(String deafaultTopic) {
        this.deafaultTopic = deafaultTopic;
    }

    @Bean
    KafkaTemplate<String,Object> kafkaProducer(){
        Map<String,Object> producerMaps = producerProps();
        ProducerFactory<String,Object> producerFactory =
                new DefaultKafkaProducerFactory<>(producerMaps,new StringSerializer(),new JsonSerializer<>());
        KafkaTemplate<String,Object> kafkaTemplate = new KafkaTemplate<>(producerFactory);
        kafkaTemplate.setDefaultTopic(deafaultTopic);
        return kafkaTemplate;
    }

    private Map<String, Object> producerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.LINGER_MS_CONFIG, lineMs);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
//        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
//        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,valueSerializer);
        return props;
    }

}
